#include "sql/operator/update_physical_operator.h"
#include "common/log/log.h"
#include "storage/trx/trx.h"
#include "sql/stmt/update_stmt.h"
#include "common/lang/mutex.h"
#include "storage/record/record.h"
#include "storage/table/table.h"

RC UpdatePhysicalOperator::open(Trx *trx)
{
  if (children_.empty()) {
    return RC::SUCCESS;
  }

  unique_ptr<PhysicalOperator> &child = children_[0];

  RC rc = child->open(trx);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to open child operator: %s", strrc(rc));
    return rc;
  }

  trx_ = trx;

  while (OB_SUCC(rc = child->next())) {
    Tuple *tuple = child->current_tuple();
    if (nullptr == tuple) {
      LOG_WARN("failed to get current record: %s", strrc(rc));
      return rc;
    }

    RowTuple *row_tuple = static_cast<RowTuple *>(tuple);
    Record   &record    = row_tuple->record();
    Record new_record;
    new_record.copy_data(record.data(), record.len());
    new_record.set_rid(record.rid());
    records_.emplace_back(std::move(new_record));
  }

  child->close();
  
  const TableMeta  table_meta = table_->table_meta();
  const FieldMeta *field_meta = table_meta.field(field_.c_str());

  // 先收集记录再删除
  // 记录的有效性由事务来保证，如果事务不保证删除的有效性，那说明此事务类型不支持并发控制，比如VacuousTrx
  for (Record &record : records_) {
    // rc = trx_->delete_record(table_, record);
    const char *value_data = value_->data();
    char *data = record.data();
    if (!value_->is_null()) {
      if (value_->attr_type() == AttrType::CHARS) {
        memcpy(data + field_meta->offset(), value_data, (size_t)value_->length()+1);
        LOG_INFO("field: %s, desc: %s, src: %s, length: %u",field_meta->name() ,data + field_meta->offset(), value_data, (size_t)value_->length()+1);
      } else {
        memcpy(data + field_meta->offset(), value_data, (size_t)value_->length());
      }
    } else {
      data[field_meta->offset()+field_meta->len()] = value_->is_null();
    }
    rc = table_->update_record(record.rid(),record);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to update record: %s", strrc(rc));
      return rc;
    }
  }

  return RC::SUCCESS;
}

RC UpdatePhysicalOperator::next()
{
  return RC::RECORD_EOF;
}

RC UpdatePhysicalOperator::close(){ return RC::SUCCESS; }
